package club.drguo.mapreduce.inverseindex;

import java.io.IOException;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 倒排索引的第一个步骤
 * 
 * @author guo
 *
 */
// club.drguo.mapreduce.inverseindex.InverseIndexStepOne
public class InverseIndexStepOne {
	// <输入：LongWritable（那一行的起始偏移量），Text（那一行的所有数据）
	// 输出：Text（hello-->a.txt），LongWritable（出现的次数）>
	public static class InverseIndexStepOneMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
		private Text k = new Text();
		private LongWritable v = new LongWritable();

		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			// 拿到第一行数据
			String line = value.toString();
			// 切分
			String[] words = StringUtils.split(line, " ");
			/**
			 * 就是先知道切分的word来自哪个文件（word--x.txt）
			 */
			// 获取本次调用传递进来的数据所在的文件信息，先要获取所属切片信息
			FileSplit inputSplit = (FileSplit) context.getInputSplit();
			// 从切片信息中获取到文件路径及文件名
			String fileName = inputSplit.getPath().getName();

			// 输出 kv对 < hello--a.txt , 1>
			for (String word : words) {
				k.set(word + "--" + fileName);
				v.set(1);
				context.write(k, v);
				// context.write(new Text(word + "--" + fileName), new
				// LongWritable(1));
			}

		}

	}

	public static class InverseIndexStepOneReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
		private LongWritable v = new LongWritable();

		// <hello-->a.txt ,{1,1,1...}>
		@Override
		protected void reduce(Text key, Iterable<LongWritable> values, Context context)
				throws IOException, InterruptedException {

			// 遍历values进行累加
			long count = 0;
			for (LongWritable value : values) {
				count += value.get();
			}
			v.set(count);
			context.write(key, v);// key还是原来的key(hello-->a.txt)
		}
	}

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();

		Job job_stepOne = Job.getInstance(conf);

		job_stepOne.setJarByClass(InverseIndexStepOne.class);

		job_stepOne.setMapperClass(InverseIndexStepOneMapper.class);
		job_stepOne.setReducerClass(InverseIndexStepOneReducer.class);

		job_stepOne.setOutputKeyClass(Text.class);
		job_stepOne.setOutputValueClass(LongWritable.class);

		FileInputFormat.setInputPaths(job_stepOne, new Path(args[0]));
		// 先判断一下，如果输出目录已存在则删除
		FileSystem fileSystem = FileSystem.get(conf);
		Path output = new Path(args[1]);
		if (fileSystem.exists(output)) {
			fileSystem.delete(output, true);// true递归删除
		}
		FileOutputFormat.setOutputPath(job_stepOne, output);

		System.exit(job_stepOne.waitForCompletion(true) ? 0 : 1);

	}

}